package com.thrift.client.annotation;

import com.thrift.client.config.TTSocket;
import com.thrift.client.config.ThriftClientConnectPoolFactory;
import com.thrift.client.utils.SpringUtil;
import com.thrift.client.utils.ZkUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.pool.impl.GenericObjectPool;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TMultiplexedProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cglib.proxy.Enhancer;
import org.springframework.cglib.proxy.MethodInterceptor;
import org.springframework.cglib.proxy.MethodProxy;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Component;

import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * Client 代理
 *
 * @author 何锦涛--CodeSleep
 * @since 2022/9/4 20:45
 */
@Component
public class ThriftClientProxy implements MethodInterceptor {

    private static final Logger logger = LoggerFactory.getLogger(ThriftClientProxy.class);

    // Tsocket
    private TTSocket ttSocket;
    // 协议对象
    private TProtocol tProtocol;
    // 连接池
    ThriftClientConnectPoolFactory poolFactory;
    // 客户端连接对象
    private Object client;
    // 服务名
    private String serviceName;
    // zk 连接
    private static final ZkUtil ZkClient = new ZkUtil("192.168.xxx.x:2181, 192.168.xxx.x:2182, 192.168.xxx.x:2183");
    // RPC 所有相关节点公用的顶级父节点
    private static final String THRIFT_SERVER_PREFIX = "/thrift/";
    // 需要增强的 bean
    private Object bean;

    public Object bind(Object bean, Class clazz, TTSocket tSocket, String serviceName, ThriftClientConnectPoolFactory poolFactory) throws Exception {
        this.bean = bean;
        this.serverListening(THRIFT_SERVER_PREFIX.concat(serviceName));
        this.serviceName = serviceName;
        this.ttSocket = tSocket;
        this.tProtocol = tProtocol;
        this.poolFactory = poolFactory;
        TTransport tTransport;
        try {
            tTransport = tSocket.open();
        } catch (Exception e) {
            logger.error("Thrift-Server 服务不在线，连接失败");
            return null;
        }
        TBinaryProtocol tProtocol = new TBinaryProtocol(tTransport);
        Constructor<Object> constructor = clazz.getDeclaredConstructor(TProtocol.class);
        TMultiplexedProtocol tMultiplexedProtocol = new TMultiplexedProtocol(tProtocol, serviceName);
        this.client = constructor.newInstance(tMultiplexedProtocol);
        // Enhancer 字节码增强创建动态代理对象
        Enhancer enhancer = new Enhancer();
        enhancer.setSuperclass(this.client.getClass());
        enhancer.setCallback(this);
        return enhancer.create(new Class[]{TProtocol.class}, new Object[]{this.tProtocol});
    }

    @Override
    public Object intercept(Object o, Method method, Object[] objects, MethodProxy methodProxy) throws Throwable {
        try {
            if (client == null) {
                logger.error("代理方法中获取的 Client 为空，可能是 Thrift Server 端不在线，注入 Client 失败");
            }
            return methodProxy.invoke(client, objects);
        } catch (Exception e) {
            logger.error("Thrift Client 调用 RPC 发生异常：{}", e.toString());
            e.printStackTrace();
            return null;
        } finally {
            // 将一个用完的对象返还给对象池
            poolFactory.returnConnection(ttSocket);
        }
    }

    /**
     * 注册 zk 监听器
     *
     * @param nodePath 待监听的路径
     */
    private void serverListening(String nodePath) {
        // 注册 zk 目录监听器，获取变化的节点数据
        ZkClient.registerPathChildrenListener(nodePath, (curatorClient, event) -> {
            ChildData childData = event.getData();
            if (childData == null)
                return;
            switch (event.getType()) {
                case CHILD_ADDED:               // 新增 RPC 节点
                    logger.info(String.format("path children add children node now, 监听的节点 %s 正在新增节点 %s", nodePath, childData.getPath()));
                    reBindClient(childData.getPath().substring(childData.getPath().lastIndexOf("/") + 1));
                    break;
                case CHILD_REMOVED:             // 减少 RPC 节点
                    logger.info(String.format("path children delete children node now, 监听的节点 %s 正在移除节点 %s", nodePath, childData.getPath()));
                    removePool(childData.getPath().substring(childData.getPath().lastIndexOf("/") + 1));
                    break;
                case CONNECTION_LOST:           // RPC 节点丢失
                    logger.info(String.format("path children connect is lost now, 监听的节点 %s 丢失了 %s 的连接", nodePath, childData.getPath()));
                    break;
                case CONNECTION_RECONNECTED:    // 节点重连
                    logger.info(String.format("path children connect is reconnected now, 监听的节点 %s 重新获取到了 %s 的连接", nodePath, childData.getPath()));
                    break;
                default:
                    break;
            }
        });
    }

    /**
     * 节点被移除，将连接从 map 中移除
     */
    private void removePool(String substring) {

        Map<String, List<ThriftClientConnectPoolFactory>> poolListMap = ThriftClientBeanPostProcessor.connectPoolListMap;
        List<ThriftClientConnectPoolFactory> poolList = poolListMap.get(serviceName);
        String host = StringUtils.substringBefore(substring, ":");
        int port = Integer.parseInt(StringUtils.substringAfter(substring, ":"));
        ThriftClientConnectPoolFactory rmvPool = null;
        for (ThriftClientConnectPoolFactory pool : poolList) {
            if (host.equals(pool.host) && pool.port == port) {
                rmvPool = pool;
                break;
            }
        }
        if (rmvPool != null) {
            rmvPool.invalidateObject(this.ttSocket);
            poolList.remove(rmvPool);
            poolListMap.put(substring, poolList);
        }
    }

    /**
     * 重新绑定连接
     */
    private void reBindClient(String substring) throws Exception {
        String host = StringUtils.substringBefore(substring, ":");
        int port = Integer.parseInt(StringUtils.substringAfter(substring, ":"));
        Map<String, List<ThriftClientConnectPoolFactory>> poolListMap = ThriftClientBeanPostProcessor.connectPoolListMap;
        List<ThriftClientConnectPoolFactory> poolList = poolListMap.get(serviceName);
        ThriftClientConnectPoolFactory poolFactory;
        if (poolList == null || poolList.size() == 0) {
            poolList = new ArrayList<>();
            GenericObjectPool.Config config = new GenericObjectPool.Config();
            config.maxActive = 3;
            config.maxWait = 3000L;
            // 创建一个池工厂对象
            poolFactory = new ThriftClientConnectPoolFactory(config, host, port, serviceName);
            poolList.add(poolFactory);
            poolListMap.put(serviceName, poolList);
            logger.info("在后置处理器中关于 {} 的连接池已创建", host + "/" + port + " " + serviceName);
        } else {
            boolean isContans = false;
            ThriftClientConnectPoolFactory emptyPool = null;
            for (ThriftClientConnectPoolFactory pool : poolList) {
                if ("".equals(pool.host) && pool.port == 0) {
                    emptyPool = pool;
                    continue;
                }
                String ip = pool.host + ":" + pool.port;
                isContans = substring.equals(ip);
            }
            if (emptyPool != null) {
                poolList.remove(emptyPool);
            }
            if (!isContans) {
                GenericObjectPool.Config config = new GenericObjectPool.Config();
                config.maxActive = 3;
                config.maxWait = 3000L;
                ThriftClientConnectPoolFactory poolFactoryCur = new ThriftClientConnectPoolFactory(config, host, port, serviceName);
                poolList.add(poolFactoryCur);
            }
            poolFactory = ThriftClientBeanPostProcessor.getLeastConnPollFactory(poolList);
            poolListMap.put(serviceName, poolList);
            logger.info("在后置处理器有多个服务端，关于 {} 的连接池已创建", host + "/" + port + " " + serviceName);
        }
        TTSocket tSocket = poolFactory.getConnect();
        ApplicationContext applicationContext = SpringUtil.getApplicationContext();
        Object bean = applicationContext.getBean(this.bean.getClass());
        Field[] fields = bean.getClass().getDeclaredFields();
        for (Field field : fields) {
            if (field.getAnnotation(ThriftClient.class) != null) {
                TTransport tTransport = tSocket.open();
                TBinaryProtocol tProtocol = new TBinaryProtocol(tTransport);
                Class clazz = field.getType();
                Constructor<Object> constructor = clazz.getDeclaredConstructor(TProtocol.class);
                TMultiplexedProtocol tMultiplexedProtocol = new TMultiplexedProtocol(tProtocol, serviceName);
                this.ttSocket = tSocket;
                this.client = constructor.newInstance(tMultiplexedProtocol);
                this.tProtocol = tProtocol;
                this.poolFactory = poolFactory;
                Enhancer enhancer = new Enhancer();
                enhancer.setSuperclass(this.client.getClass());
                enhancer.setCallback(this);
                Object clientProxy = enhancer.create(new Class[]{TProtocol.class}, new Object[]{this.tProtocol});
                field.set(bean, clientProxy);
            }
        }
    }
}

